package com.demoworld;

import org.elasticsearch.action.bulk.*;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.junit.Test;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Created by francis on 2017/6/14.
 */
public class main {
private static TransportClient client;

static {
    Settings esSettings = Settings.builder()
            .put("cluster.name", "elasticsearch") //设置ES实例的名称
            .put("client.transport.sniff", true) //自动嗅探整个集群的状态，把集群中其他ES节点的ip添加到本地的客户端列表中
            .build();
    client = new PreBuiltTransportClient(esSettings);//初始化client较老版本发生了变化，此方法有几个重载方法，初始化插件等。
    //此步骤添加IP，至少一个，其实一个就够了，因为添加了自动嗅探配置
    try {
        client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));
    } catch (UnknownHostException e) {
        e.printStackTrace();
    }

}

    /**
     * 查看集群信息
     */
    @Test
    public void testInfo() {
        List<DiscoveryNode> nodes = client.connectedNodes();
        for (DiscoveryNode node : nodes) {
            System.out.println(node.getHostAddress());
            System.out.println(node.getName());
        }
    }

@Test
public void add() throws InterruptedException {
    Map<String, Object> infoMap = new HashMap<String, Object>();
    infoMap.put("name", "广告信息11 test1");
    infoMap.put("title", "我的广告 test1 java net python");
    infoMap.put("createTime", new Date());
    infoMap.put("count", 1251);
    IndexResponse indexResponse = client.prepareIndex("test1", "info1", "2545").setSource(infoMap).execute().actionGet();
    System.out.println("id:" + indexResponse.getId());
    client.close();
}

    @Test
    public void searchById() {
        GetResponse response = client.prepareGet("test", "info", "1")
                .execute().actionGet();
        System.out.println("response.getId():" + response.getId());
        System.out.println("response.getSourceAsString():" + response.getSourceAsString());
        client.close();
    }

//范围查询
@Test
public void query() throws Exception {
    //range查询
    QueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("count").gt(50);
    SearchResponse searchResponse = client.prepareSearch("test")
            .setTypes("info")
            .setQuery(rangeQueryBuilder)
            .addSort("count", SortOrder.DESC)
            .setSize(20)
            .execute()
            .actionGet();
    SearchHits hits = searchResponse.getHits();
    System.out.println("查到记录数：" + hits.getTotalHits());
    SearchHit[] searchHists = hits.getHits();
    if (searchHists.length > 0) {
        for (SearchHit hit : searchHists) {
            String name = (String) hit.getSource().get("name");
            System.out.format("name:%s ,count :%s \n", name, hit.getSource().get("count"));
        }
    }
}

//相等查询
@Test
public void query1() throws Exception {
    //term查询
    QueryBuilder queryBuilder = QueryBuilders.termQuery("count", 1022);
    SearchResponse searchResponse = client.prepareSearch("test")
            .setTypes("info")
            .setQuery(queryBuilder)
            .addSort("count", SortOrder.DESC)
            .setSize(20)
            .execute()
            .actionGet();
    SearchHits hits = searchResponse.getHits();
    System.out.println("查到记录数：" + hits.getTotalHits());
    SearchHit[] searchHists = hits.getHits();
    if (searchHists.length > 0) {
        for (SearchHit hit : searchHists) {
            String name = (String) hit.getSource().get("name");
            System.out.format("name:%s ,count :%s \n", name, hit.getSource().get("count"));
        }
    }
}

//模糊查询
@Test
public void query2() throws Exception {
    //term查询
    QueryBuilder queryBuilder = QueryBuilders.fuzzyQuery("title", "java");
    SearchResponse searchResponse = client.prepareSearch("test")
            .setTypes("info")
            .setQuery(queryBuilder)
            .addSort("count", SortOrder.DESC)
            .setSize(20)
            .execute()
            .actionGet();
    SearchHits hits = searchResponse.getHits();
    System.out.println("查到记录数：" + hits.getTotalHits());
    SearchHit[] searchHists = hits.getHits();
    if (searchHists.length > 0) {
        for (SearchHit hit : searchHists) {
            String name = (String) hit.getSource().get("name");
            System.out.format("name:%s ,count :%s \n", name, hit.getSource().get("count"));
        }
    }
}

    //删除指定id的数据
    @Test
    public void testDelete() {
        DeleteResponse response = client.prepareDelete("test", "info", "100")
                .get();
        String index = response.getIndex();
        String type = response.getType();
        String id = response.getId();
        long version = response.getVersion();
        System.out.println(index + " : " + type + ": " + id + ": " + version);
    }

    @Test
    public void testUpdate() throws Exception {
        UpdateRequest updateRequest = new UpdateRequest();
        updateRequest.index("test");
        updateRequest.type("info");
        updateRequest.id("100");
        updateRequest.doc(XContentFactory.jsonBuilder()
                .startObject()
                // 对没有的字段添加, 对已有的字段替换
                .field("gender", "male")
                .endObject());
        UpdateResponse response = client.update(updateRequest).get();

        // 打印
        String index = response.getIndex();
        String type = response.getType();
        String id = response.getId();
        long version = response.getVersion();
        System.out.println(index + " : " + type + ": " + id + ": " + version);
    }

    @Test
    public void testUpdate_UsingScript() throws Exception {
        // 使用Script对象进行更新
        UpdateResponse response = client.prepareUpdate("test", "info", "100")
                .setScript(new Script("ctx._source.gender = \"aaa\""))
                .get();
        System.out.println(response.getIndex());
        client.close();
    }

    @Test
    public void testUpdate_UsingScript1() throws Exception {
        UpdateRequest updateRequest = new UpdateRequest("test", "info", "100")
                .script(new Script("ctx._source.gender=\"male\""));
        UpdateResponse response = client.update(updateRequest).get();
        System.out.println(response.toString());
        client.close();
    }

    // 测试upsert方法
    @Test
    public void testUpsert() throws Exception {
        // 设置查询条件, 查找不到则添加生效
        IndexRequest indexRequest = new IndexRequest("test", "info", "100")
                .source(XContentFactory.jsonBuilder()
                        .startObject()
                        .field("name", "qergef")
                        .field("gender", "malfdsae")
                        .endObject());
        // 设置更新, 查找到更新下面的设置
        UpdateRequest upsert = new UpdateRequest("test", "info", "100")
                .doc(XContentFactory.jsonBuilder()
                        .startObject()
                        .field("user", "wenbronk")
                        .endObject())
                .upsert(indexRequest);

        UpdateResponse response = client.update(upsert).get();
        System.out.println(response.toString());
    }

    /**
     * 测试multi get api
     * 从不同的index, type, 和id中获取
     */
    @Test
    public void testMultiGet() {
        MultiGetResponse multiGetResponse = client.prepareMultiGet()
                .add("test", "info", "100")
                .add("test1", "info1", "101")
                .add("test2", "info2", "102")
                .get();

        for (MultiGetItemResponse itemResponse : multiGetResponse) {
            GetResponse response = itemResponse.getResponse();
            if (response != null && response.isExists()) {
                String sourceAsString = response.getSourceAsString();
                System.out.println(sourceAsString);
            }
        }
    }

    /**
     * bulk 批量执行
     * 一次查询可以update 或 delete多个document
     */
    @Test
    public void testBulk() throws Exception {
        BulkRequestBuilder bulkRequest = client.prepareBulk();
        bulkRequest.add(client.prepareIndex("test", "info", "1")
                .setSource(XContentFactory.jsonBuilder()
                        .startObject()
                        .field("count", "1022")
                        .endObject()));
        bulkRequest.add(client.prepareIndex("test1", "info1", "101")
                .setSource(XContentFactory.jsonBuilder()
                        .startObject()
                        .field("user", "kimchy")
                        .field("postDate", new Date())
                        .field("message", "another post")
                        .endObject()));
        BulkResponse response = bulkRequest.get();

        Iterator<BulkItemResponse> ite = response.iterator();
        while (ite.hasNext()) {
            System.out.println(ite.next().getId());
        }
    }

    //使用监听器 查询
    @Test
    public void testBulkProcessor() throws Exception {
        // 创建BulkPorcessor对象
        BulkProcessor bulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
            public void beforeBulk(long paramLong, BulkRequest paramBulkRequest) {
                System.out.println("start indexing query");
            }

            // 执行出错时执行
            public void afterBulk(long paramLong, BulkRequest paramBulkRequest, Throwable paramThrowable) {
                System.out.println(paramThrowable.toString());
            }

            public void afterBulk(long paramLong, BulkRequest paramBulkRequest, BulkResponse paramBulkResponse) {
                Iterator<BulkItemResponse> ite = paramBulkResponse.iterator();
                while (ite.hasNext()) {
                    System.out.println(ite.next().getId());
                }
            }
        })
                // 1w次请求执行一次bulk
                .setBulkActions(10000)
                // 1gb的数据刷新一次bulk
                .setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
                // 固定5s必须刷新一次
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                // 并发请求数量, 0不并发, 1并发允许执行
                .setConcurrentRequests(1)
                // 设置退避, 100ms后执行, 最大请求3次
                .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
                .build();

        // 这个请求报参数错误， 目前没有找到原因
//        bulkProcessor.add(new IndexRequest("test", "info", "1"));
        bulkProcessor.add(new DeleteRequest("test1", "info1", "101"));

        // 关闭
        bulkProcessor.awaitClose(10, TimeUnit.MINUTES);
        client.close();
        System.out.println("over");
        // 或者
//        bulkProcessor.close();
    }

}
